package com.atguigu.flink.chapter6;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
import org.apache.flink.util.Collector;

import java.util.HashMap;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2020/12/22 10:05
 */
public class Flink05_Project_Order {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        SingleOutputStreamOperator<OrderEvent> orderStream = env
          .readTextFile("input/OrderLog.csv")
          .map(new MapFunction<String, OrderEvent>() {
              @Override
              public OrderEvent map(String line) throws Exception {
                  String[] datas = line.split(",");
                  return new OrderEvent(Long.valueOf(datas[0]), datas[1], datas[2], Long.valueOf(datas[3]));

              }
          });

        SingleOutputStreamOperator<TxEvent> txStream = env
          .readTextFile("input/ReceiptLog.csv")
          .map(new MapFunction<String, TxEvent>() {
              @Override
              public TxEvent map(String line) throws Exception {
                  String[] datas = line.split(",");
                  return new TxEvent(datas[0], datas[1], Long.valueOf(datas[2]));

              }
          });

        ConnectedStreams<OrderEvent, TxEvent> connectedStream = orderStream.connect(txStream);

        connectedStream
          .keyBy("txId", "txId")
          .process(new KeyedCoProcessFunction<String, OrderEvent, TxEvent, String>() {
              // txId->OrderEvent
              HashMap<String, OrderEvent> orderMap = new HashMap<>();
              HashMap<String, TxEvent> txMap = new HashMap<>();

              @Override
              public void processElement1(OrderEvent value, Context ctx, Collector<String> out) throws Exception {
                  // 判断对应的交易数据是否已经存在, 如果存在对账成功, 否则把订单数据存入map集合
                  if(txMap.containsKey((value.getTxId()))){
                      String currentKey = value.getTxId();
                      out.collect(currentKey + " 对账成功!!!");
                      txMap.remove(currentKey);
                  }else{
                      orderMap.put(value.getTxId(), value);
                  }
              }

              @Override
              public void processElement2(TxEvent value, Context ctx, Collector<String> out) throws Exception {
                    if(orderMap.containsKey(value.getTxId())){
                        out.collect(value.getTxId() + " 对账成功!!!");
                        orderMap.remove(value.getTxId());
                    }else{
                        txMap.put(value.getTxId(), value);
                    }
              }

          })
          .print();
        env.execute();
    }
}
